package com.yytek.am.core.runtime;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import com.alibaba.fastjson.JSON;
import com.yytek.am.constants.MessageType;
import com.yytek.am.core.msg.InputDataMessage;
import com.yytek.am.core.msg.MqConfigUpdateMessage;
import com.yytek.am.core.msg.OutputDataMessage;
import com.yytek.am.core.rule.EventRule;
import com.yytek.am.core.rule.EventRuleClear;
import com.yytek.am.core.rule.EventRuleGenerate;
import com.yytek.common.client.MqClientFactory;
import com.yytek.common.client.base.IMqClient;
import com.yytek.common.client.base.MessageCallBack;
import com.yytek.common.client.base.MqConfig;
import com.yytek.common.client.base.MqMessage;

import java.util.Arrays;
import java.util.List;

/**
 * 订阅告警原始数据 广播告警事件
 */
public class EventSourceSinkActor extends AbstractActor {
    private IMqClient mqClient;
    private ActorRef eventManagerActor;
    private RuleManager ruleManager;
    /**
     * 内置监听告警规则topic
     */
    public final List<String> builtInTopicList = Arrays.asList(MessageType.EventRule.EVENT_RULE,
            MessageType.EventRule.EVENT_RULE_GENERATE, MessageType.EventRule.EVENT_RULE_CLEAR);

    public static Props props(ActorRef eventManagerActor, RuleManager ruleManager) {
        return Props.create(EventSourceSinkActor.class, () -> new EventSourceSinkActor(eventManagerActor, ruleManager));
    }

    public EventSourceSinkActor(ActorRef eventManagerActor, RuleManager ruleManager) {
        this.eventManagerActor = eventManagerActor;
        this.ruleManager = ruleManager;
    }

    @Override
    public void preStart() {

    }

    @Override
    public void postStop() {
        if (mqClient != null) {
            mqClient.close();
        }
    }

    /**
     * mqtt订阅消息到达
     * @param topic
     * @param mqMessage
     */
    private void onMessageArrived(String topic, MqMessage mqMessage) {
        byte[] payload = mqMessage.getPayload();
        int topicLen = topic.length();
        boolean builtIn = false;
        // 内置修改删除告警定义和规则
        for (String builtInTopic : builtInTopicList) {
            int builtInTopicLen = builtInTopic.length();
            if (topicLen == builtInTopicLen && topic.startsWith(builtInTopic.substring(0, builtInTopicLen - 1))) {
                int type = Integer.valueOf(topic.substring(topicLen - 1)).intValue();
                switch (builtInTopic) {
                    case MessageType.EventRule.EVENT_RULE:
                        ruleManager.updateEventRule(type, JSON.parseObject(payload, EventRule.class));
                        break;
                    case MessageType.EventRule.EVENT_RULE_GENERATE:
                        ruleManager.updateEventRuleGenerate(type, JSON.parseObject(payload, EventRuleGenerate.class));
                        break;
                    case MessageType.EventRule.EVENT_RULE_CLEAR:
                        ruleManager.updateEventRuleClear(type, JSON.parseObject(payload, EventRuleClear.class));
                        break;
                    default:
                        break;
                }
                builtIn = true;
                break;
            }
        }
       //如果不是告警定义修改topic，是输入数据
        if (!builtIn) {
            // 告警源数据发送到输入数据eventMananger处理
            eventManagerActor.tell(new InputDataMessage(JSON.parseObject(payload, InputDataMessage.InputData.class)), this.getSelf());
        }
    }

    private void onMqConfigUpdateMessage(MqConfigUpdateMessage message) {
        if (mqClient != null) {
            mqClient.close();
        }
        MqConfig config = message.mqConfig;
        config.setClientId(config.getClientId() + "_" + System.nanoTime());
        mqClient = MqClientFactory.getInstance().createMQTTClient(config);
        mqClient.open();
        //mqtt设置消息回调接收
        mqClient.setMessageCallBack(new MessageCallBack() {
            @Override
            public void onPublish(String topic, MqMessage mqMessage, String destination) {
                onMessageArrived(topic, mqMessage);
            }
        });
        // 内置订阅mqtt消息
        for (String builtInTopic : builtInTopicList) {
            mqClient.subscribe(builtInTopic);
        }
        // 告警源订阅mqtt消息
        for (String topic : message.topicList) {
            mqClient.subscribe(topic);
        }
    }

    private void onOutputDataMessage(OutputDataMessage message) {
        //推送告警事件
        mqClient.publish(MessageType.AM_ALARM_EVENT, message.outputData);
    }

    @Override
    public Receive createReceive() {
        //收到MQTT配置
        return receiveBuilder().match(MqConfigUpdateMessage.class, this::onMqConfigUpdateMessage)
                //输出数据到mqtt发送
                .match(OutputDataMessage.class, this::onOutputDataMessage).build();
    }

}
